package com.dbms;

import com.dbms.catalog.Catalog;
import com.dbms.common.Column;
import com.dbms.common.Schema;
import com.dbms.common.Tuple;
import com.dbms.common.Type;
import com.dbms.execution.execution.*;
import com.dbms.storage.BufferPoolManager;
import com.dbms.storage.DiskManager;
import com.dbms.transaction.LogManager;

import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Random;

/**
 * 基于规则的查询优化器（RBO）之“谓词下推（Predicate Pushdown）”功能演示程序
 * 核心目标：验证谓词下推对查询性能的提升，对比“先连接后过滤”与“先过滤后连接”的效率差异
 *
 * 测试场景设计：
 * - 大表 `events`（事件表）：存储大量事件记录（10万条），通过`e_type_id`关联事件类型
 * - 小表 `event_types`（事件类型表）：存储少量事件类型（100种），包含一个“罕见事件类型”
 * - 查询需求：查找“罕见事件类型”对应的所有事件记录（结果极少，仅5条）
 *
 * 对比两种执行计划：
 * 1. 【未优化】先对两表做完整Hash Join，再对海量连接结果过滤 → 开销极大
 *    执行树：Filter → HashJoin → SeqScan(events)、SeqScan(event_types)
 * 2. 【谓词下推优化】先过滤小表`event_types`得到极少结果（1条），再与大表Join → 开销骤降
 *    执行树：HashJoin → SeqScan(events)、(Filter → SeqScan(event_types))
 *
 * 预期结果：优化后性能提升数个数量级，体现谓词下推在“小表过滤+大表连接”场景的价值
 */
public class PredicatePushdownTest {

    // --- 测试参数配置 ---
    private static final int EVENT_TYPE_COUNT = 100;    // 事件类型表记录数（小表，100种类型）
    private static final int EVENT_COUNT = 100_000;     // 事件表记录数（大表，10万条事件）
    private static final String RARE_EVENT_NAME = "SYSTEM_CRASH"; // 待查询的罕见事件类型名
    private static final String DB_FILE_NAME = "optimizer_test.db"; // 测试用数据库文件
    private static final String LOG_FILE_NAME = "optimizer_test.log"; // 测试用日志文件

    /**
     * 程序主入口：执行“谓词下推”性能对比的完整流程
     * @param args 命令行参数（未使用）
     * @throws IOException 磁盘IO异常（如文件创建、数据写入失败）
     */
    public static void main(String[] args) throws IOException {
        System.out.println("============== RBO 谓词下推性能对比演示开始 ==============");

        // --- 1. 数据准备阶段：初始化环境、创建表结构、插入测试数据 ---
        // 此阶段耗时不计入性能对比，核心是构建“大表+小表+罕见数据”的测试场景
        System.out.println("\n--- 1. 数据准备阶段 ---");
        long setupStartTime = System.nanoTime(); // 纳秒级计时，确保精度

        // 清理旧测试文件，避免历史数据干扰
        new File(DB_FILE_NAME).delete();
        new File(LOG_FILE_NAME).delete();

        // 初始化数据库核心组件（存储、缓冲、元数据）
        DiskManager diskManager = new DiskManager(DB_FILE_NAME); // 物理文件读写管理
        LogManager logManager = new LogManager(LOG_FILE_NAME);   // 事务日志管理
        BufferPoolManager bpm = new BufferPoolManager(200, diskManager, logManager); // 200页缓存，减少IO
        Catalog catalog = new Catalog(bpm, true); // 元数据目录，管理表结构/索引

        // -------------------------- 创建小表 event_types（事件类型表） --------------------------
        Schema eventTypeSchema = new Schema(List.of(
                new Column("et_id", Type.INTEGER, true),   // 事件类型ID（主键，唯一标识）
                new Column("et_name", Type.VARCHAR, 50)    // 事件类型名（如"SYSTEM_CRASH"）
        ));
        // 执行建表操作
        new CreateTableExecutor(catalog, "event_types", eventTypeSchema).next();

        // -------------------------- 创建大表 events（事件表） --------------------------
        Schema eventSchema = new Schema(List.of(
                new Column("e_id", Type.INTEGER, true),       // 事件ID（主键）
                new Column("e_type_id", Type.INTEGER),        // 关联的事件类型ID（外键，关联event_types.et_id）
                new Column("e_data", Type.VARCHAR, 100)       // 事件详情（冗余字段，模拟真实数据）
        ));
        // 执行建表操作
        new CreateTableExecutor(catalog, "events", eventSchema).next();

        // -------------------------- 插入事件类型数据（小表） --------------------------
        Catalog.TableMetadata eventTypeTable = catalog.getTable("event_types");
        // 插入前99种普通事件类型（ID 1~99）
        for (int i = 1; i < EVENT_TYPE_COUNT; i++) {
            Tuple tuple = new Tuple(List.of(i, "EventType_" + i), eventTypeSchema);
            new InsertExecutor(bpm, eventTypeTable, Collections.singletonList(tuple)).next();
        }
        // 插入第100种“罕见事件类型”（ID=EVENT_TYPE_COUNT，名称=RARE_EVENT_NAME）
        Tuple rareTuple = new Tuple(List.of(EVENT_TYPE_COUNT, RARE_EVENT_NAME), eventTypeSchema);
        new InsertExecutor(bpm, eventTypeTable, Collections.singletonList(rareTuple)).next();

        // -------------------------- 插入事件数据（大表） --------------------------
        Catalog.TableMetadata eventTable = catalog.getTable("events");
        Random random = new Random();
        int rareEventCount = 0; // 统计罕见事件的实际插入数量

        for (int i = 1; i <= EVENT_COUNT; i++) {
            int eventTypeId;
            // 刻意控制罕见事件的概率：每2万条事件插入1条（共5条，确保结果集极小）
            if (i % (EVENT_COUNT / 5) == 0) {
                eventTypeId = EVENT_TYPE_COUNT; // 关联罕见事件类型
                rareEventCount++;
            } else {
                // 普通事件：关联1~99的普通类型ID
                eventTypeId = random.nextInt(EVENT_TYPE_COUNT - 1) + 1;
            }
            // 创建事件记录并插入
            Tuple tuple = new Tuple(List.of(i, eventTypeId, "Event_Data_" + i), eventSchema);
            new InsertExecutor(bpm, eventTable, Collections.singletonList(tuple)).next();
        }

        // 刷新缓冲池所有数据到磁盘，确保后续查询基于稳定的物理数据
        bpm.flushAllPages();

        // 打印数据准备结果
        long setupEndTime = System.nanoTime();
        System.out.printf("数据准备完成！共插入 %d 事件类型和 %d 事件记录。\n", EVENT_TYPE_COUNT, EVENT_COUNT);
        System.out.printf("其中 '%s' 类型的事件共 %d 条。\n", RARE_EVENT_NAME, rareEventCount);
        System.out.printf("准备耗时: %.3f ms\n", (setupEndTime - setupStartTime) / 1_000_000.0);
        System.out.println("--------------------------------------------------\n");

        // --- 2. 性能测试阶段：对比未优化与谓词下推优化的执行耗时 ---
        System.out.println("--- 2. 查询性能测试阶段 ---");
        System.out.printf("    查询: SELECT * FROM events JOIN event_types ON event_types.et_id = events.e_type_id WHERE event_types.et_name = '%s';\n", RARE_EVENT_NAME);

        // -------------------------- 测试1：未优化计划（先Join后Filter） --------------------------
        double unoptimizedMs = testWithoutPredicatePushdown(bpm, catalog);
        System.out.printf("\n--> 测试1: [未优化 - Filter after Join]  耗时: %.3f 毫秒\n", unoptimizedMs);

        // -------------------------- 测试2：优化计划（谓词下推，先Filter后Join） --------------------------
        double optimizedMs = testWithPredicatePushdown(bpm, catalog);
        System.out.printf("--> 测试2: [谓词下推优化 - Filter before Join] 耗时: %.3f 毫秒\n", optimizedMs);

        // --- 3. 结果总结：计算性能提升倍数，验证优化效果 ---
        System.out.println("\n--- 性能对比总结 ---");
        System.out.printf("    未优化计划耗时:      %.3f ms (基准)\n", unoptimizedMs);
        System.out.printf("    谓词下推优化后耗时:   %.3f ms\n", optimizedMs);
        // 计算性能提升倍数（避免除零异常）
        if (optimizedMs > 0 && unoptimizedMs > optimizedMs) {
            System.out.printf("    性能提升: 优化后的速度是未优化前的 %.2f 倍！\n", unoptimizedMs / optimizedMs);
        } else {
            System.out.println("    性能提升巨大！");
        }

        System.out.println("\n============== 演示结束 ==============");
    }

    /**
     * 测试【未优化】的执行计划：先做Hash Join，再过滤连接结果
     * 执行流程：SeqScan(events) + SeqScan(event_types) → HashJoin → Filter(找罕见事件)
     * 问题：Join结果为10万条，过滤后仅5条，大量无效计算
     * @param bpm 缓冲池管理器
     * @param catalog 元数据目录
     * @return 执行耗时（毫秒）
     */
    private static double testWithoutPredicatePushdown(BufferPoolManager bpm, Catalog catalog) {
        System.out.println("    执行计划 (未优化): Filter -> HashJoin -> SeqScan(events), SeqScan(event_types)");

        // 1. 底层扫描：全表扫描大表events和小表event_types
        Executor eventsScan = new SeqScanExecutor(bpm, catalog, "events"); // 扫描10万条事件
        Executor eventTypesScan = new SeqScanExecutor(bpm, catalog, "event_types"); // 扫描100种类型

        // 2. Hash Join：对两表做完整连接，生成10万条连接结果
        Schema joinSchema = getJoinedSchema(catalog); // 连接后的结果集结构（合并两表列）
        Predicate joinPredicate = new Predicate(1, 0); // 连接条件：events.e_type_id(索引1) = event_types.et_id(索引0)
        Executor hashJoin = new HashJoinExecutor(eventsScan, eventTypesScan, joinPredicate, joinSchema);

        // 3. Filter：在10万条连接结果中过滤“罕见事件类型”（低效！）
        // 连接后Schema列顺序：e_id(0)、e_type_id(1)、e_data(2)、et_id(3)、et_name(4)
        Predicate filterPredicate = new Predicate(4, "=", RARE_EVENT_NAME); // 过滤条件：et_name(索引4) = 罕见事件名
        Executor finalPlan = new FilterExecutor(filterPredicate, hashJoin);

        // 执行计划并计时
        return executeAndMeasure(finalPlan);
    }

    /**
     * 测试【谓词下推优化】的执行计划：先过滤小表，再做Hash Join
     * 执行流程：SeqScan(events) + (SeqScan(event_types) → Filter) → HashJoin
     * 优势：小表过滤后仅1条结果，Join时仅需匹配大表中5条数据，开销骤降
     * @param bpm 缓冲池管理器
     * @param catalog 元数据目录
     * @return 执行耗时（毫秒）
     */
    private static double testWithPredicatePushdown(BufferPoolManager bpm, Catalog catalog) {
        System.out.println("    执行计划 (已优化): HashJoin -> SeqScan(events), (Filter -> SeqScan(event_types))");

        // 1. 底层扫描：全表扫描大表events，全表扫描小表event_types
        Executor eventsScan = new SeqScanExecutor(bpm, catalog, "events"); // 扫描10万条事件
        Executor eventTypesScan = new SeqScanExecutor(bpm, catalog, "event_types"); // 扫描100种类型

        // 2. 谓词下推：先过滤小表event_types，仅保留1条罕见事件类型（关键优化！）
        // 小表Schema列顺序：et_id(0)、et_name(1)，过滤条件：et_name(索引1) = 罕见事件名
        Predicate filterPredicate = new Predicate(1, "=", RARE_EVENT_NAME);
        Executor filteredEventTypes = new FilterExecutor(filterPredicate, eventTypesScan); // 过滤后仅1条结果

        // 3. Hash Join：用1条小表结果与大表Join，仅匹配5条事件（高效！）
        Schema joinSchema = getJoinedSchema(catalog); // 连接后的结果集结构
        Predicate joinPredicate = new Predicate(1, 0); // 连接条件：events.e_type_id(1) = event_types.et_id(0)
        Executor finalPlan = new HashJoinExecutor(eventsScan, filteredEventTypes, joinPredicate, joinSchema);

        // 执行计划并计时
        return executeAndMeasure(finalPlan);
    }

    /**
     * 辅助方法：执行指定查询计划，计算执行耗时（毫秒）并统计匹配记录数
     * @param executor 待执行的查询计划执行器
     * @return 执行耗时（毫秒）
     */
    private static double executeAndMeasure(Executor executor) {
        long startTime = System.nanoTime(); // 开始计时

        // 执行查询计划：初始化→遍历结果→关闭资源
        executor.open(); // 初始化执行器（如加载数据页、初始化哈希表）
        int rowCount = 0; // 统计最终匹配的记录数
        while (executor.next() != null) { // 逐行获取结果，直到无更多数据
            rowCount++;
        }
        executor.close(); // 关闭执行器，释放资源（如解锁页、释放缓存）

        long endTime = System.nanoTime(); // 结束计时

        // 打印记录数，返回耗时（纳秒转毫秒：除以1e6）
        System.out.printf("    (查询完成，共找到 %d 条匹配记录)", rowCount);
        return (endTime - startTime) / 1_000_000.0;
    }

    /**
     * 辅助方法：构建两表连接后的结果集Schema（合并events和event_types的所有列）
     * 列顺序：events的所有列 → event_types的所有列
     * @param catalog 元数据目录（用于获取表结构）
     * @return 连接后的Schema
     */
    private static Schema getJoinedSchema(Catalog catalog) {
        List<Column> finalColumns = new ArrayList<>();
        // 先添加大表events的所有列（e_id, e_type_id, e_data）
        finalColumns.addAll(catalog.getTable("events").schema.getColumns());
        // 再添加小表event_types的所有列（et_id, et_name）
        finalColumns.addAll(catalog.getTable("event_types").schema.getColumns());
        return new Schema(finalColumns);
    }
}